package com.gome.han.bigdata.spark.core.rdd.operation.demo

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author Hanpeng
 * @date 2021/1/14 20:38
 * @description:
 */
object Demo1 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 1. 获取原始数据：时间戳，     省份，城市，用户，广告
    //                 1516609143869 1   4    44   15
    //统计出每一个省份每个广告被点击数量排行的 Top3
    val lines: RDD[String] = sc.textFile("in/agent.log");
    //分成24个区 ((省份,广告),1)
    val mapRdd: RDD[((String, String), Int)] = lines.map(line => {
      val datas: Array[String] = line.split(" ");

      ((datas(1), datas(4)), 1)
    })
    // 3. 将转换结构后的数据，进行分组聚合
    val reduceRDD: RDD[((String, String), Int)] = mapRdd.reduceByKey(_+ _)
    // 4. 将聚合的结果进行结构的转换
    //    ( ( 省份，广告 ), sum ) => ( 省份, ( 广告, sum ) )
    val newMapRDD = reduceRDD.map{
      case ( (prv, ad), sum ) => {
        (prv, (ad, sum))
      }
    }
    // 5. 将转换结构后的数据根据省份进行分组
    //    ( 省份, 【( 广告A, sumA )，( 广告B, sumB )】 )
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = newMapRDD.groupByKey()
    // 6. 将分组后的数据组内排序（降序），取前3名
    val resultRDD = groupRDD.mapValues(
      iter => {
        //(x,y)=> x + y
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
      }
    )
    // 7. 采集数据打印在控制台
    resultRDD.collect().foreach(println)
    sc.stop()
  }
}
